package cn.doitedu.sql;

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.table.functions.AggregateFunction;

public class _18_CustomFunction_Aggregate {

    public static void main(String[] args) {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        StreamTableEnvironment tenv = StreamTableEnvironment.create(env);

        tenv.executeSql(
                " create table score_kfk(                                    "+
                        "       user_id int,                                    "+
                        "       course string,                                  "+
                        " 	    score double                                     "+
                        "  ) with (                                              "+
                        "       'connector' = 'kafka',                           "+
                        "       'topic' = 'score-test',                            "+
                        "       'properties.bootstrap.servers' = 'doitedu:9092', "+
                        "       'properties.group.id' = 'g003',                  "+
                        "       'scan.startup.mode' = 'latest-offset',           "+
                        "       'value.format' = 'json',                         "+
                        "       'value.fields-include' = 'EXCEPT_KEY'            "+
                        " )                                                      "
        );


        //
        tenv.createTemporaryFunction("my_avg", MyAvgFunction.class);
        tenv.executeSql("select  course,  my_avg(score) from score_kfk group by course").print();





    }


    public static class AvgAccumulator{
        public int count;
        public double scoreSum;

        public AvgAccumulator(int count, double scoreSum) {
            this.count = count;
            this.scoreSum = scoreSum;
        }

        public AvgAccumulator() {
        }
    }


    public static class MyAvgFunction extends AggregateFunction<Double,AvgAccumulator>  {

        /**
         * 获取最终结果
         */
        @Override
        public Double getValue(AvgAccumulator accumulator) {
            return accumulator.scoreSum/accumulator.count;
        }

        /**
         * 创建中间聚合数据结构
         */
        @Override
        public AvgAccumulator createAccumulator() {
            return new AvgAccumulator(0,0d);
        }

        /**
         * 单步聚合逻辑
         */
        public void accumulate(AvgAccumulator acc, Double score) {
           acc.count = acc.count+1;
           acc.scoreSum = acc.scoreSum+score;

        }

        /**
         * 当输入数据有-u/+u,就会调用该方法，来对之前的聚合结果做修正
         */
        public void retract(AvgAccumulator acc, Double score) {
           acc.count = acc.count-1;
           acc.scoreSum = acc.scoreSum - score;
        }


        /**
         * 多个局部聚合中间数据结构，合并成一个中间数据结构
         */
        public void merge(AvgAccumulator accumulator, Iterable<AvgAccumulator> it) {

            for (AvgAccumulator acc : it) {
                accumulator.scoreSum += acc.scoreSum;
                accumulator.count += acc.count;
            }

        }

        /**
         * 重置中间数据结构
         */
        public void resetAccumulator(AvgAccumulator acc) {
            acc.count = 0;
            acc.scoreSum = 0;
        }
    }
}
